package cn.doitedu.rtdw.data_sync;


import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/13
 * @Desc: 学大数据，上多易教育
 *   将业务库中的用户注册信息表数据，实时同步到hbase
 *
 *   mysql中的表：  ums_member
 *   hbase中的表：  dim_user_info
 **/
public class Job01_用户注册信息表同步 {
    public static void main(String[] args) {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 对数据源创建逻辑映射表
        tenv.executeSql("create table ums_member_mysql (\n" +
                "  id bigint                          \n" +
                "  ,member_level_id bigint            \n" +
                "  ,username string                   \n" +
                "  ,password string                   \n" +
                "  ,nickname string                   \n" +
                "  ,phone string                      \n" +
                "  ,status int                        \n" +
                "  ,create_time timestamp(3)          \n" +
                "  ,icon string                       \n" +
                "  ,gender int                        \n" +
                "  ,birthday date                     \n" +
                "  ,city string                       \n" +
                "  ,job string                        \n" +
                "  ,personalized_signature string     \n" +
                "  ,source_type int                   \n" +
                "  ,integration int                   \n" +
                "  ,growth int                        \n" +
                "  ,luckey_count int                  \n" +
                "  ,history_integration int           \n" +
                "  ,modify_time timestamp(3)          \n" +
                "  ,primary key (id)  not enforced    \n" +
                ")WITH (                       \n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = 'doitedu',   \n" +
                "    'port' = '3306',          \n" +
                "    'username' = 'root',      \n" +
                "    'password' = 'root',      \n" +
                "    'database-name' = 'realtimedw',\n" +
                "    'table-name' = 'ums_member'\n" +
                ")");


        // 对目标存储创建逻辑映射表
        tenv.executeSql(
                " create table dim_user_info_hbase(       "+
                        " 	user_name string,                      "+
                        "   f  row<                                " +
                        "          id  bigint                      "+
                        " 	       ,member_level_id bigint         "+
                        "          ,password string                "+
                        "          ,nickname string                "+
                        "          ,phone string                   "+
                        "          ,status int                     "+
                        "          ,create_time timestamp(3)       "+
                        "          ,icon string                    "+
                        "          ,gender int                     "+
                        "          ,birthday date                  "+
                        "          ,city string                    "+
                        "          ,job string                     "+
                        "          ,personalized_signature string  "+
                        "          ,source_type int                "+
                        "          ,integration int                "+
                        "          ,growth int                     "+
                        "          ,luckey_count int               "+
                        "          ,history_integration int        "+
                        "          ,modify_time timestamp(3)       "+
                        " 	>                                      "+
                        " ) WITH (                                 "+
                        "  'connector' = 'hbase-2.2',              "+
                        "  'table-name' = 'dim_user_info',         "+
                        "  'zookeeper.quorum' = 'doitedu:2181'     "+
                        " )                                        "
        );




        // 写一个sql：  insert into 目标逻辑表   select   from  数据源逻辑表
        tenv.executeSql("INSERT INTO dim_user_info_hbase\n" +
                "SELECT\n" +
                "  username                      \n" +
                " ,row(                          \n" +
                "       id                       \n" +
                "      ,member_level_id         \n" +
                "      ,password                 \n" +
                "      ,nickname                 \n" +
                "      ,phone                    \n" +
                "      ,status                   \n" +
                "      ,create_time              \n" +
                "      ,icon                     \n" +
                "      ,gender                   \n" +
                "      ,birthday                 \n" +
                "      ,city                     \n" +
                "      ,job                      \n" +
                "      ,personalized_signature  \n" +
                "      ,source_type              \n" +
                "      ,integration              \n" +
                "      ,growth                   \n" +
                "      ,luckey_count             \n" +
                "      ,history_integration     \n" +
                "      ,modify_time ) as f      \n" +
                "FROM ums_member_mysql");




    }
}
